package com.idroc.coelacanth.service.client;

import java.util.Map;
import java.util.Random;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.coelacanth.core.dht.NodeActionObserver.NodeAction;
import com.idroc.coelacanth.core.membership.Endpoint;
import com.idroc.coelacanth.core.message.MessageDispatchHandler;
import com.idroc.coelacanth.service.MessageService;
import com.idroc.coelacanth.service.ThreadPoolService;
import com.idroc.coelacanth.service.servant.ClientRegisterMessage;
import com.idroc.coelacanth.service.servant.NodeActionNoticeMessage;
import com.idroc.coelacanth.service.servant.ServantSetRequestMessage;
import com.idroc.coelacanth.service.servant.ServantSetResponseMessage;

public class StroageProxy {
	
	private static final Logger log = LoggerFactory.getLogger(StroageProxy.class);
	
	private Endpoint endpoint = null;
	
	private Endpoint regTarget = null;
	
	private MessageService msgService = null;
	
	private final SortedMap<Integer, Endpoint> servants = new TreeMap<Integer, Endpoint>();
	
	private final Map<Endpoint, StorageServantBroker> servantBrokers = new ConcurrentHashMap<Endpoint, StorageServantBroker>();
	
	private boolean runningServantFetchTask = true;

	public StroageProxy(Endpoint ep, Endpoint quote, MessageService ms) {
		
		this.endpoint = ep;
		this.msgService = ms;
		registerMessageHandler();
		buildServants(quote);
	}
	
	private void registerMessageHandler() {
		
		MessageDispatchHandler dispatcher = msgService.getServerMessageFactory().getMessageDispatcher();
		// 处理"存储服侍器集合响应消息"
		dispatcher.addUpMessageHandler(ServantSetResponseMessage.class, new ServantSetResponseMessageHandler());
		// 处理"节点动作通知消息"
		dispatcher.addUpMessageHandler(NodeActionNoticeMessage.class, new NodeActionNoticeMessageHandler());
	}
	
	/**
	 * 建立服务侍候器集合
	 * @param quote -- 引用端点
	 */
	public void buildServants(final Endpoint quote) {
		
		Runnable task = new Runnable() {

			private int tryCount = 100; // try 100 times
			
			public void run() { 
				if (runningServantFetchTask && tryCount > 0) {
					tryCount--;
					ServantSetRequestMessage msg = new ServantSetRequestMessage();
					msg.setFrom(endpoint);
					msgService.sendMessage(quote, msg);
					ThreadPoolService.getMaintainSchduledThreadPool().schedule(this, 3, TimeUnit.SECONDS);
				} else {
					runningServantFetchTask = true;
				}
			}
		}; 
		ThreadPoolService.getMaintainSchduledThreadPool().schedule(task, 0, TimeUnit.SECONDS);
	} 
	
	/**
	 * 向服务侍候器注册
	 */
	private void registerToServant() {
		 
		Random random = new Random();
		int next = random.nextInt(servants.size());
		regTarget = servants.get(Integer.valueOf(next));
		ClientRegisterMessage msg = new ClientRegisterMessage();
		msg.setFrom(endpoint);
		log.debug("向服务侍候器[{}]注册", regTarget);
		msgService.sendMessage(regTarget, msg);
	}
	
	/**
	 * 查找服务侍候器代理
	 * @param key -- 编号
	 * @return -- 服务侍候器代理
	 */
	public StorageServantBroker lookup(int key) {  
		
		log.debug("使用编号[{}]查找服务侍候器代理, servants[{}]", key, servants);
		
		Integer k = Integer.valueOf(key);
		Endpoint s = servants.get(k);
		if (null == s) {
			SortedMap<Integer, Endpoint> tails = servants.tailMap(k);
			if (tails.isEmpty()) {
				s = servants.get(servants.firstKey());
			} else {
				s = servants.get(tails.firstKey());
			}
		}
		log.debug("servantBrokers[{}]", servantBrokers);
		StorageServantBroker broker = servantBrokers.get(s); 
		return broker;
	}
	 	
	/**
	 * 存储服侍器集合响应消息 处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-16 下午05:58:27
	 */
	private class ServantSetResponseMessageHandler  extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {

			ServantSetResponseMessage msg = (ServantSetResponseMessage)e.getMessage();
			log.debug("存储服侍器集合响应消息 处理器,消息:{}", msg); 
			
			runningServantFetchTask = false; //停止存储服侍器查询任务
			
			Map<Integer, Endpoint> s = msg.getServants();
			for (Integer key : s.keySet()) {
				Endpoint ep = s.get(key);
				StorageServantBroker broker = new StorageServantBroker();
				broker.setAction(NodeAction.OPEN);
				broker.setServant(ep);
				servantBrokers.put(ep, broker);
			}
			servants.putAll(s);
			log.debug("当前存储服侍器集合[{}]", servants);
			registerToServant(); // 注册客户端
		}
	}
	
	/**
	 * 节点动作通知消息 处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-17 下午01:38:23
	 */
	private class NodeActionNoticeMessageHandler  extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {

			NodeActionNoticeMessage msg = (NodeActionNoticeMessage)e.getMessage();
			log.debug("节点动作通知消息 处理器,消息:{}", msg); 
			
			Integer no = Integer.valueOf(msg.getNodeNo());
			Endpoint ep = msg.getActor();
			
			log.debug("no[{}], ep[{}]", no, ep);
			
			if (msg.getAction() == NodeAction.INSERT) {
				
				// 创建新的节点
				StorageServantBroker broker = new StorageServantBroker();
				broker.setAction(NodeAction.INSERT);
				broker.setServant(ep);
				
				// 建立备份关系
				servantBrokers.put(ep, broker);
				StorageServantBroker next = lookup(no.intValue());
				next.setBackup(broker);
				
				
				servants.put(no, ep); // 放入环中
				
				log.debug("新的侍候器加入当前服务侍候器集合[{}]", servants);
				log.debug("新的侍候器[{}]分担侍候器[{}]的压力", broker, next);
				
			} else {
				log.debug("异常消息, 未知动作, 没处理消息");
			}
		}
	}
}
